﻿#region LicenseHeader

// Copyright 2012 The Trustees of Indiana University.  All rights reserved.
// 
// The Indiana University licenses this file to you under the Apache License, 
// Version 2.0 (the "License"); you may not use this file except in compliance 
// with the License.  You may obtain a copy of the License at 
// 
// http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software 
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
// See the License for the specific language governing permissions and 
// limitations under the License.
// 
// @author Thilina Gunarathne (tgunarat@indiana.edu)

#endregion

using System;
using System.Collections.Generic;
using System.IO;
using System.Runtime.CompilerServices;
using AzureMRCore.DataTypes;
using AzureMRCore.Primitives;

namespace AzureMRCore.InputFormat
{
    public class CachedDoubleMatrixFormat : IInputFormat<IntKey, MatrixValue<double>>, ICachedInputFormat, IInputFormat
    {
        private int _counter;
        private Boolean _isKeyed;
        private double[][] _values;
        private int _height, _width, _row, _rowOffset;
        private long _size;
        private bool _isRead;
        private MatrixValue<double> _value;

        #region ICachedInputFormat Members

        public long GetSize()
        {
            return _size;
        }

        public ICachedInputFormat GetCachedCopy()
        {
            CachedDoubleMatrixFormat newCachedFormat = new CachedDoubleMatrixFormat();
            newCachedFormat.Init(_values, _size, _width, _height, _row, _rowOffset);
            return newCachedFormat;
        }

        #endregion

        private void Init(double[][] values, long size, int vectorLength,
                          int numVectors, int row, int rowOffset)
        {
            _values = values;
            _size = size;
            _width = vectorLength;
            _height = numVectors;
            _row = row;
            _rowOffset = rowOffset;
        }

        #region IInputFormat<IntKey,double[]> Members

        public void Init(string inputFile)
        {
            BinaryReader reader = new BinaryReader(File.Open(inputFile, FileMode.Open, FileAccess.Read, FileShare.Read));
            FileInfo file = new FileInfo(inputFile);
            _size = file.Length;
            _height = reader.ReadInt32();
            _width = reader.ReadInt32();
            _row = reader.ReadInt32();
            _rowOffset = reader.ReadInt32();
            _isKeyed = reader.ReadBoolean();

            _values = new double[_height][];

            //_isKeyed = _reader.ReadBoolean();
            ReadAll(reader);
            reader.Close();
        }

        public void Init(IDictionary<int, BCastDataProduct> inputRecords, int numMapTasks)
        {
            /*var values = from item in inputRecords
                         orderby ((MatrixValue<double>)item.Value.Value).RowBlock ascending
                         select item.Value;
            var lastValue = ((MatrixValue<double>) values.Last().Value);
*/
            // sanity check
            BCastDataProduct lastRecord;
            if ((numMapTasks != inputRecords.Count) | (!(inputRecords.TryGetValue(numMapTasks - 1, out lastRecord))))
                throw new Exception("Error in fetched BCast data. NummapTasks != Bcast record count.");

            var lastValue = (MatrixValue<double>) lastRecord.Value;
            _height = lastValue.BlockHeight*numMapTasks; //lastValue.RowOffset + lastValue.BlockHeight;
            _rowOffset = 0;
            _row = 0;
            _width = lastValue.Width;
            _values = new double[_height][];

            //TODO currently support only single values
            for (int j = 0; j < numMapTasks; j++)
            {
                var value = (MatrixValue<double>) inputRecords[j].Value;
                for (int i = 0; i < value.BlockHeight; i++)
                {
                    _values[value.RowOffset + i] = value.Matrix[i];
                }
            }
        }


        public IValue FromBytes(byte[] values)
        {
            _value = new MatrixValue<double>();
            _value.FromBytes(values);
            return _value;
        }

        public List<KeyValuePair<IntKey, MatrixValue<double>>> GetAllPairs()
        {
            if (_value == null)
                _value = MatrixValue<double>.GetInstance(_values, _height,
                                                         _width, _row, _rowOffset);

            var list = new List<KeyValuePair<IntKey, MatrixValue<double>>>
                           {
                               new KeyValuePair<IntKey, MatrixValue<double>>(IntKey.GetInstance(0),
                                                                             _value)
                           };
            return list;
        }

        public IntKey GetCurrentKey()
        {
            IntKey intKey = new IntKey {Key = 0};
            return intKey;
        }

        public MatrixValue<double> GetCurrentValue()
        {
            return _value ?? (_value = MatrixValue<double>.GetInstance(_values, _height, _width, _row, _rowOffset));
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public bool Next()
        {
            if (!_isRead)
            {
                _isRead = true;
                return true;
            }

            return false;
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public bool HasNext()
        {
            return !(_isRead);
        }

        public void Close()
        {
        }

        #endregion

        private double[][] ReadAll(BinaryReader reader)
        {
            while (_counter < _height)
            {
                double[] vector;
                if (_isKeyed)
                {
                    vector = new double[_width + 1];
                    int key = reader.ReadInt32();
                    vector[_width] = key;
                }
                else
                {
                    vector = new double[_width];
                }
                for (var i = 0; i < _width; i++)
                {
                    vector[i] = reader.ReadDouble();
                }
                _values[_counter] = vector;
                _counter++;
            }
            return _values;
        }
    }
}